package com.atguigu.gmall.realtime.test;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author Felix
 * @date 2023/12/20
 * 该案例演示了Flink的普通连接
 *                                  左表                    右表
 *          内连接             onCreateAndWrite        onCreateAndWrite
 *          左外连接           onReadAndWrite          onCreateAndWrite
 *          右外连接           onCreateAndWrite        onReadAndWrite
 *          全外连接           onReadAndWrite          onReadAndWrite
 */
public class Flink03_SQL_BaseJoin {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从指定的网络端口读取员工数据
        SingleOutputStreamOperator<Emp> empDS = env
            .socketTextStream("hadoop102", 8888)
            .map(lineStr -> {
                String[] fieldArr = lineStr.split(",");
                return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
            });
        //TODO 4.将员工流转换为动态表
        //Table empTable = tableEnv.fromDataStream(empDS);
        tableEnv.createTemporaryView("emp",empDS);

        //TODO 5.从指定的网络端口读取部门数据
        SingleOutputStreamOperator<Dept> deptDS = env
            .socketTextStream("hadoop102", 8889)
            .map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Dept(Integer.valueOf(fieldArr[0]), fieldArr[1], Long.valueOf(fieldArr[2]));
                }
            );
        //TODO 6.将部门流转换为动态表
        tableEnv.createTemporaryView("dept",deptDS);

        //TODO 7.内连接
        //注意：如果使用普通的内外连接，底层会为参与连接的两张表各自维护一个状态，用于存放参与连接的两张表的数据
        //默认情况下，状态永不会失效
        //在生产环境中，需要手动设置状态的失效时间
        //tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10))
        //  不能单独设置某张表的失效时间，只能统一设置
        //  这里的时间指定是处理时间
        //  一般在设置时间的时候考虑两方面因素：传输延迟  +  业务上的滞后关系
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e join dept d on e.deptno = d.deptno").print();

        //TODO 8.左外连接
        //如果左表数据先到，右表数据后到，会产生3条数据
        //当左表数据到来的时候，会生成     +I [左表    null]
        //当右表数据到来的时候，会先生成   -D [左表    null]
        //再生成                       +I [左表     右表]
        //这样的动态表转换为的流称之为回撤流
        //如果将这样的动态表数据插入到kafka主题，kafka主题会接收到3条消息
        //  [左表     null]
        //  null
        //  [左表     右表]
        //当从kafka主题中读取数据的时候，如果使用的是FlinkSQL的方式，会自动处理空消息
        //如果使用的是FlinkAPI的方式从kafka主题中读取数据，默认的SimpleStringSchema是处理不了空消息的
        //需要自定义反序列化器处理空消息
        //另外：对于重复消息，在DWS中会详细讲解去重操作
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno").print();

        //TODO 9.右外连接
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e right join dept d on e.deptno = d.deptno").print();

        //TODO 10.全外连接
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e full join dept d on e.deptno = d.deptno").print();

        //TODO 11.将左外连接的结果写到kafka主题
        //11.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE emp_dept (\n" +
            "  empno integer,\n" +
            "  ename string,\n" +
            "  deptno integer,\n" +
            "  dname string,\n" +
            "  PRIMARY KEY (empno) NOT ENFORCED\n" +
            ") WITH (\n" +
            "  'connector' = 'upsert-kafka',\n" +
            "  'topic' = 'first',\n" +
            "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
            "  'key.format' = 'json',\n" +
            "  'value.format' = 'json'\n" +
            ")");
        tableEnv.executeSql("insert into emp_dept select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno");

    }
}
